package com.lzcu.maoyan.mapreduce.mapreduce.mapper;

import com.lzcu.maoyan.mapreduce.mapreduce.reduce.MovieMonthCountReduce;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieMonthCountWritable;
import com.lzcu.maoyan.mapreduce.service.PhoenixMapper;
import com.lzcu.maoyan.mapreduce.service.impl.PhoenixMapperImpl;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.format.DateTimeParseException;
import java.io.IOException;
import java.time.temporal.ChronoField;
import com.lzcu.maoyan.mapreduce.mapreduce.unit.MovieTypeWritable;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.ArrayList;

/**
 * @author WangLiang
 * @date `2024/5/8`
 */
public class MovieMonthCountMapper extends Mapper<NullWritable, MovieMonthCountWritable, Text, Text>  {
    @Override
    protected void map(NullWritable key, MovieMonthCountWritable value, Context context)
            throws IOException, InterruptedException {
        String release = value.getReleaseDate();
        if (null != release) {
            String releaseDate = value.getReleaseDate().replaceAll("-", "");
            String year = releaseDate.substring(0, 4);
            String month = releaseDate.substring(4, 6);
            if ("2024".equals(year)) {
                try {
                    long movieBoxOffice = new PhoenixMapperImpl()
                            .getMovieBoxOffice(value.getSourceMovieName(),
                                    year + "-" + month);
                    if (movieBoxOffice != 0) {
                        context.write(new Text(year + "-" + month),
                                new Text(value.getSourceMovieName() + "\001" +
                                movieBoxOffice));
                    }
                } catch (Exception e){
                    e.printStackTrace();
                }
            }
        }
    }
}
